package study.core.rdd.operator

import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子之键值操作combineByKey操作
 * 最通用的对 key-value 型 spark.core.rdd 进行聚集操作的聚集函数，允许用户返回值的类型与输入不一致
 *
 * @author zh
 * @date 2021/5/15 10:47
 */
object TestCombineByKey {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("rddMap")
    val context = new SparkContext(sparkConf)

    // 准备集合
    val list = List(("a", 1), ("b", 2), ("a", 3), ("b", 4), ("a", 5))
    // 创建RDD
    val rdd = context.makeRDD(list, 2)

    // 第一个参数是更改数据的格式
    // 第二个参数 分区内操作   分区内根据key分别进行累加和计数  元组中第一个元素为累加，第二个元素为计数
    // 第三个参数  分区间操作  分区内根据key分别进行累加和计数  元组中第一个元素为累加，第二个元素为计数
    val combineResult = rdd.combineByKey(
      (v => (v, 1)),
      (x: (Int, Int), y) => {
        (x._1 + y, x._2 + 1)
      },
      (x: (Int, Int), y: (Int, Int)) => {
        (x._1 + y._1, x._2 + y._2)
      })

    // 求平均值
    val result = combineResult.mapValues((res) => {
      res._1 / res._2
    })

    result.collect().foreach(println)

    context.stop()

    //(b,3)
    //(a,3)

  }

}
